package cn.itcast.mr;

import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;


/**
 * Author itcast
 * Date 2021/10/15 15:14
 * Desc TODO
 */
public class Hdfs {

    private FileSystem fileSystem = null;
    @Before
    public void getFileSystem() throws IOException {
        //添加模拟操作
        System.setProperty("HADOOP_USER_NAME","root");
        //设置配置 fs.defaultFS 文件系统
        Configuration config = new Configuration();
        //指的是namenode的访问端口
        config.set("fs.defaultFS","hdfs://192.168.88.161:8020");
        //获取指定的文件系统
        fileSystem = FileSystem.get(config);
        //打印输出
        System.out.println(fileSystem);
    }


    public void getFileList() throws IOException {
        // listFiles 从根路径下遍历所有的文件
        RemoteIterator<LocatedFileStatus> fileLists = fileSystem
                .listFiles(new Path("/"), true);
        // 如果存在就获取打印输出路径
        while(fileLists.hasNext()){
            LocatedFileStatus fileStatus = fileLists.next();
            System.out.println(fileStatus.getPath().toString());
        }
        // 关闭文件系统
        fileSystem.close();
    }


    public void mkdirs() throws IOException {
        //创建文件夹 mkdirs
        boolean flag = fileSystem.mkdirs(new Path("/user/hive"));
        if(flag){
            System.out.println("创建文件夹成功");
        }else{
            System.out.println("创建文件夹失败");
        }
        fileSystem.close();
    }

    public void downloadFile() throws IOException {
        // 文件系统打开hdfs上的文件输入数据流
        FSDataInputStream inputStream = fileSystem.open(new Path("/tmp/a"));
        // 实例化本地文件输出流
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        FSDataOutputStream fsDataOutputStream = localFileSystem.create(new Path("file:///E:\\hello.txt"));
        // 通过IOUtils工具类copy输入输出流
        IOUtils.copy(inputStream,fsDataOutputStream);
        // 关闭输入输出流
        IOUtils.closeQuietly(inputStream);
        IOUtils.closeQuietly(fsDataOutputStream);
    }


    public void upload() throws IOException {
        //通过copyFromLocalFile 拷贝本地数据到文件系统上
        fileSystem.copyFromLocalFile(new Path("file:///E:\\names.txt"),new Path("/tmp/"));
        fileSystem.close();
    }

    /**
     * 输入流本地文件系统
     * 输出到 HDFS 文件系统，指定文件系统的名称
     */
    @Test
    public void merge() throws IOException {
        // 文件系统create合并的文件的外部输入流
        FSDataOutputStream dos = fileSystem.create(new Path("/user/merge.dat"));
        FSDataInputStream inputStream = null;
        // 文件系统获取本地文件系统
        LocalFileSystem localFileSystem = FileSystem.getLocal(new Configuration());
        //FSDataInputStream inputStream = localFileSystem.open(new Path("file:///E:\\tmp"));
        // 通过listStatus获取本地文件状态数组
        FileStatus[] fileStatus = localFileSystem.listStatus(new Path("file:///E:\\tmp"));
        // 遍历文件状态
        for (FileStatus status : fileStatus) {
            inputStream = localFileSystem.open(status.getPath());
            IOUtils.copy(inputStream,dos);
        }
        // 本地文件系统打开文件状态路径
        // 考本本地输入流到外部输入流
        // 关闭连接
        IOUtils.closeQuietly(fileSystem);
        IOUtils.closeQuietly(localFileSystem);
        IOUtils.closeQuietly(dos);
        IOUtils.closeQuietly(inputStream);
    }

}
